package shardkv

import (
	"time"
)
import "6.824/raft"

func (kv *ShardKV) Get(args *GetArgs, reply *GetReply) {
	// Your code here.
	opArgs := OpCommand{
		Type:     "Get",
		Key:      args.Key,
		ClientId: args.ClientId,
		CmdId:    args.CmdId,
	}
	opReply := Reply{}
	kv.HandleRequest(&opArgs, &opReply)
	reply.Err, reply.Value = opReply.Err, opReply.Value
}

func (kv *ShardKV) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	opArgs := OpCommand{
		Type:     args.Op,
		Key:      args.Key,
		Value:    args.Value,
		ClientId: args.ClientId,
		CmdId:    args.CmdId,
	}
	opReply := Reply{}
	kv.HandleRequest(&opArgs, &opReply)
	reply.Err = opReply.Err
}

func (kv *ShardKV) HandleRequest(args *OpCommand, reply *Reply) {
	kv.mu.Lock()
	switch args.Type {
	case "Get":
		kv.logger.Printf("HandleRequest: 开始处理Get请求（ClientId=%d, CmdId=%d, Key=%s）", args.ClientId%10000, args.CmdId, args.Key)
	case "Put":
		kv.logger.Printf("HandleRequest: 开始处理Put请求（ClientId=%d, CmdId=%d, Key=%s, Value=%s）", args.ClientId%10000, args.CmdId, args.Key, args.Value)
	case "Append":
		kv.logger.Printf("HandleRequest: 开始处理Append请求（ClientId=%d, CmdId=%d, Key=%s, Value=%s）", args.ClientId%10000, args.CmdId, args.Key, args.Value)
	}

	shardId := key2shard(args.Key)
	curConfig := kv.curConfig
	shardState := kv.Shards[shardId].State
	//当前副本不是切片拥有者 或者拥有，但是状态不是ready或neednotify都不允许访问
	if curConfig.Shards[shardId] != kv.gid {
		// 分片不属于当前组
		reply.Err = ErrWrongGroup
		kv.logger.Printf("HandleRequest: 分片不属于当前组（ShardId=%d, 实际所属Group=%d, 当前Group=%d）", shardId, curConfig.Shards[shardId], kv.gid)
		kv.mu.Unlock()
		return
	} else if shardState != Ready && shardState != NeedNotify {
		// 分片状态不合法（非Ready/NeedNotify）
		reply.Err = ErrWrongGroup
		kv.logger.Printf("HandleRequest: 分片状态不合法（ShardId=%d, State=%d，需Ready或NeedNotify）", shardId, shardState)
		kv.mu.Unlock()
		return
	}

	if args.Type != "Get" && kv.checkDuplicated(args) {
		kv.logger.Printf("HandleRequest: 检测到重复请求（ClientId=%d, CmdId=%d），已处理过该请求", args.ClientId%10000, args.CmdId)
		kv.mu.Unlock()
		reply.Err = ErrDuplicate
		return
	}

	index, term, isLeader := kv.rf.Start(*args)
	if !isLeader {
		kv.logger.Printf("HandleRequest: 当前节点不是Leader（ClientId=%d, CmdId=%d），拒绝处理请求", args.ClientId%10000, args.CmdId)
		kv.mu.Unlock()
		reply.Err = ErrWrongLeader
		return
	}
	kv.logger.Printf("HandleRequest: 请求已提交到Raft（ClientId=%d, CmdId=%d, RaftIndex=%d, Term=%d）", args.ClientId%10000, args.CmdId, index, term)
	//用有缓冲通道来避免出现无人接收的极端情况，减少一个goroutine的浪费
	ch := make(chan *Reply, 1)
	kv.pendingChs[index] = ch
	kv.pendingOps[index] = args
	kv.mu.Unlock()

	kv.logger.Printf("请求已提交到Raft: ClientId=%d, CmdId=%d, Index=%d, Term=%d",
		args.ClientId%10000, args.CmdId, index, term)

	//最多等待时长
	timer := time.NewTimer(500 * time.Millisecond)
	defer timer.Stop()

	for {
		select {
		//在超时之前收到了命令
		case result := <-ch:
			kv.clearOld(index)
			reply.Value, reply.Err = result.Value, result.Err
			kv.logger.Printf("HandleRequest: 收到Raft应用结果（ClientId=%d, CmdId=%d, RaftIndex=%d, Err=%v）", args.ClientId%10000, args.CmdId, index, reply.Err)
			return
		//等待超时：
		case <-timer.C:
			// 清理映射表
			kv.clearOld(index)
			//可能当前节点是旧的领导者，处于少数分区，需要重新寻找新的领导者
			reply.Err = ErrWrongLeader
			kv.logger.Printf("HandleRequest: 等待超时（500ms）（ClientId=%d, CmdId=%d, RaftIndex=%d），可能已失去Leader身份", args.ClientId%10000, args.CmdId, index)
			return
		//没一百毫秒检查一下是否还是领导者且任期没有变化，任期变化即使是领导者，当前命令可能也已经不见了
		case <-time.After(100 * time.Millisecond):
			if kv.rf.StateChanged(term) {
				kv.clearOld(index)
				reply.Err = ErrWrongLeader
				kv.logger.Printf("HandleRequest: 节点状态变化（Term变化或失去Leader身份）（ClientId=%d, CmdId=%d, 旧Term=%d）", args.ClientId%10000, args.CmdId, term)
				return
			}
		}
	}
}

func (kv *ShardKV) applyCmd(msg *raft.ApplyMsg, index int) {
	op := msg.Command.(OpCommand)
	reply := Reply{}

	kv.logger.Printf("applyCmd: 开始应用Raft命令（RaftIndex=%d, Type=%s, ClientId=%d, CmdId=%d, Key=%s）", index, op.Type, op.ClientId%10000, op.CmdId, op.Key)

	//查看当前数据的切片是否是自己管理的
	shardId := key2shard(op.Key)
	//
	curConfig := kv.curConfig
	shardState := kv.Shards[shardId].State
	//当前副本不是切片拥有者 或者拥有，但是状态不是ready或neednotify都不允许访问
	if curConfig.Shards[shardId] != kv.gid {
		// 分片不属于当前组
		reply.Err = ErrWrongGroup
		kv.logger.Printf("applyCmd: 分片不属于当前组（RaftIndex=%d, ShardId=%d, 实际所属Group=%d, 当前Group=%d）", index, shardId, curConfig.Shards[shardId], kv.gid)
	} else if shardState != Ready && shardState != NeedNotify {
		// 分片状态不合法（非Ready/NeedNotify）
		reply.Err = ErrWrongGroup
		kv.logger.Printf("applyCmd: 分片状态不合法（RaftIndex=%d, ShardId=%d, State=%d，需Ready或NeedNotify）", index, shardId, shardState)
	} else {
		if op.Type == "Get" {
			val, ok := kv.Shards[shardId].DB[op.Key]
			if ok {
				reply.Err = OK
				reply.Value = val
				kv.logger.Printf("applyCmd: Get命令执行成功（RaftIndex=%d, Key=%s, Value=%s）", index, op.Key, val)
			} else {
				reply.Err = ErrNoKey
				reply.Value = ""
				kv.logger.Printf("applyCmd: Get命令执行失败（RaftIndex=%d, Key=%s，键不存在）", index, op.Key)
			}
		} else if kv.checkDuplicated(&op) {
			reply.Err = ErrDuplicate
			kv.logger.Printf("applyCmd: 检测到重复命令（RaftIndex=%d, ClientId=%d, CmdId=%d）", index, op.ClientId%10000, op.CmdId)
		} else {
			if op.Type == "Put" {
				kv.Shards[shardId].DB[op.Key] = op.Value
				kv.logger.Printf("applyCmd: Put命令执行成功（RaftIndex=%d, Key=%s, Value=%s）", index, op.Key, op.Value)
			} else {
				val, ok := kv.Shards[shardId].DB[op.Key]
				if ok {
					kv.Shards[shardId].DB[op.Key] = val + op.Value
					kv.logger.Printf("applyCmd: Append命令执行成功（RaftIndex=%d, Key=%s, 旧值=%s, 新值=%s）", index, op.Key, val, val+op.Value)
				} else {
					kv.Shards[shardId].DB[op.Key] = op.Value
					kv.logger.Printf("applyCmd: Append命令执行成功（新键）（RaftIndex=%d, Key=%s, Value=%s）", index, op.Key, op.Value)
				}
			}
			reply.Err = OK
			kv.ClientCmdSeq[op.ClientId] = op.CmdId
			kv.logger.Printf("applyCmd: 更新客户端命令序列（ClientId=%d, 最新CmdId=%d）", op.ClientId%10000, op.CmdId)
		}
	}

	ch, cexists := kv.pendingChs[index]
	oldOp, oexists := kv.pendingOps[index]
	if cexists && oexists && op.ClientId == oldOp.ClientId && op.CmdId == oldOp.CmdId {
		kv.logger.Printf("applyCmd: 发送回复到等待通道（RaftIndex=%d, ClientId=%d, CmdId=%d）", index, op.ClientId%10000, op.CmdId)
		go func() { ch <- &reply }()
	} else {
		kv.logger.Printf("applyCmd: 无匹配的等待请求，无需发送回复（RaftIndex=%d）", index)
	}
}
